package service

import (
	"context"
	"fmt"
	"prod/api/prod"
	"prod/internal/model"
	"prod/pkg/log"
	"prod/pkg/request_key"
)

type msgRetryKey struct{}

// NSQ消息恢复
func (s *Service) NSQRetry(ctx context.Context, req *prod.CommonRetryRequest) (resp *prod.DefaultResp, err error) {
	limit := int(req.Limit)
	if limit == 0 {
		limit = 1000
	}
	var messages, sendMsgs []*model.NsqMessage
	if messages, err = s.dao.FindRetryMessages(ctx, limit); err != nil {
		return
	}
	// 标记上下文为retry版本
	if len(messages) != 0 {
		requestId := request_key.GetRequestID(ctx)
		// 异步执行，不阻塞请求，防止外部请求超时时，context cancel导致后续处理失败
		go func() {
			for _, msg := range messages {
				if _, ok := processorMap[msg.Topic]; ok {
					// 消息接收重试任务
					subCtx := request_key.GenerateContextWithRequestID(requestId)
					subCtx = context.WithValue(subCtx, msgRetryKey{}, msg)
					if msgError := processorMap[msg.Topic](subCtx, msg.Payload); msgError != nil {
						log.FromContext(subCtx).Infow("nsq",
							"status_code", "Fail-Unknown",
							"method", "retry",
							"path", msg.Topic,
							"body", msg.MsgKey,
							"error", msgError.Error(),
							"stacktrace", fmt.Sprintf("msg retry error: %+v", err),
						)
					}
				} else {
					// 消息发送重试任务
					sendMsgs = append(sendMsgs, msg)
				}
			}
			if len(sendMsgs) != 0 {
				// 批量重试
				subCtx := request_key.GenerateContextWithRequestID(requestId)
				err = s.dao.Transaction(subCtx, func(subCtx context.Context) (err error) {
					err = s.dao.RetryNSQMessages(subCtx, sendMsgs)
					return
				})
			}
		}()
	}
	resp = &prod.DefaultResp{Status: true}
	return
}
